package com.dahuan.state;

import com.dahuan.bean.SensorReading;
import javafx.collections.ListChangeListener;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Collections;
import java.util.List;

public class State_OperatorState {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setParallelism( 1 );
        //TODO 基于事件时间
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );
        //水印之间的间隔（以毫秒为单位）
        env.getConfig().setAutoWatermarkInterval( 100 );

        DataStream<String> inputStream = env.socketTextStream( "localhost", 7777 );
        DataStream<SensorReading> dataStream = inputStream.map( data -> {
            String[] split = data.split( "," );
            return new SensorReading( split[0], new Long( split[1] ), new Double( split[2] ) );
        } );

        //TODO 定义一个有状态的map操作 , 统计当前分区数据个数
        SingleOutputStreamOperator<Integer> resultStream = dataStream.map( new MyCountMapper() );

        resultStream.print();

        env.execute("State_OperatorState");
    }

    //自定义MapFunction
    public static class MyCountMapper implements MapFunction<SensorReading,Integer>, ListCheckpointed<Integer> {

        //定义一个本地变量，作为算子状态
        private Integer count = 0;


        @Override
        public Integer map(SensorReading value) throws Exception {
            count++;
            return count;
        }

        @Override
        public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
            //TODO  返回一个只包含指定对象的不可变列表。
            return Collections.singletonList( count );
        }

        @Override
        public void restoreState(List<Integer> state) throws Exception {
            for (Integer num: state){
                count += num;
            }
        }
    }

}
